package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Code29Cache {
  def main(args: Array[String]): Unit = {
    /**
     * 对于MapValuesRDD在计算过程中，被重复计算了两次
     *    那么对于同一个RDD在后续的计算过程中如果被重复使用多次，那么在Spark中最好对其进行做缓存操作
     *
     * 缓存操作：
     *    方式1：cache 默认是将数据缓存到内存当中 缓存操作会将当前RDD中数据添加值缓存目标地
     *    方式2: persist 需要给定存储等级，存储等级共分为 Memory、disk、SER、replication、OFF_HEAP 五种类别
     */



    val conf: SparkConf = new SparkConf().setMaster("local").setAppName("Cache")
    val sc = new SparkContext(conf)

    val wordsRDD: RDD[String] = sc.parallelize(
      List("hello spark", "hello scala", "hello world")
    )

    val wordSplitRDD: RDD[String] = wordsRDD.flatMap(
      _.split(" ")
    )

    val groupByRDD: RDD[(String, Iterable[String])] = wordSplitRDD.groupBy(x => x)

    val countResRDD: RDD[(String, Int)] = groupByRDD
      .mapValues(
        x => {
          println("mapValues执行了...")
          x.size
        }
      )

    // 对RDD进行添加cache缓存操作
//    countResRDD.cache()

    countResRDD.persist(StorageLevel.DISK_ONLY_2)



    countResRDD
      .map(_._1.toInt)
      .saveAsTextFile("spark_code/output/wordCount")

    countResRDD
      .foreach(println)


    // 清空缓存操作
//    countResRDD.unpersist()

    while (true) {}
  }
}
